use iced::futures::future::ok;
use chrono::NaiveDateTime;
use chrono::{DateTime, Local};
use mysql_async::prelude::Queryable;
use polars::df;
use polars::prelude::*;
use polars::prelude::{DatetimeChunked, TimeUnit};


async fn read_from_mysql() -> Result<DataFrame, anyhow::Error> {
    // 1. 建立数据库连接
    let opts = mysql_async::Opts::from_url(
        "mysql://root:hdjt2022DATA!@192.168.111.205:3306/salary_chain",
        // "mysql://[用户名]:[密码]@[主机]:[端口]/[数据库名]?[参数]"
        // 常用参数示例：
        //   ssl-mode=REQUIRED       // 启用SSL加密
        //   ssl-ca=/path/to/ca.pem  // 自定义CA证书路径
        //   charset=utf8mb4         // 字符集设置
        //   useCompression=true     // 启用压缩
    )?;

    let pool = mysql_async::Pool::new(opts);
    let mut conn = pool.get_conn().await?;

    // 执行查询示例
    let sql = "select username, password, role, created_at from salary_user";

    // 2. 执行查询
    let rows: Vec<(String, String, i64, NaiveDateTime)> = conn
        .query_map(sql, |(username, password, role, created_at)| {
            (username, password, role, created_at)
        })
        .await?;

    // 3. 转换为 Polars DataFrame
    // 转换时间类型
    let usernames: Vec<String> = rows.iter().map(|(u, _, _, _)| u.to_owned()).collect();
    let passwords: Vec<String> = rows.iter().map(|(_, p, _, _)| p.to_owned()).collect();
    let role: Vec<i64> = rows.iter().map(|(_, _, r, _)| r.to_owned()).collect();

    // 转换为Polars支持的时间戳（纳秒）
    let created_at: Vec<i64> = rows
        .iter()
        .map(|(_, _, _, dt)|
            dt.and_local_timezone(Local)
                .unwrap()
                .timestamp_nanos_opt()
                .unwrap()
        )
        .collect();

    // 创建时间序列
    let created_at_series = Int64Chunked::new("创建日期".into(), &created_at)
        .into_datetime(TimeUnit::Nanoseconds, None)
        .into_series();

    Ok(DataFrame::new(vec![
        Series::new("用户名".into(), usernames).into(),
        Series::new("密码".into(), passwords).into(),
        Series::new("角色".into(), role).into(),
        created_at_series.into(),
    ])?)
}

#[tokio::main]
pub async fn main() -> Result<(), anyhow::Error> {
    let df = read_from_mysql().await?;
    println!("完整数据：\n{}", df);
    learn_polars(&df);
    Ok(())
}

fn learn_polars(df: &DataFrame) -> Result<(), Box<dyn std::error::Error>>{
    println!("数据类型：\n{:?}", df.dtypes());
    println!("数据形状：\n{:?}", df.shape());
    println!("数据前5行：\n{}", df.head(Some(5)));
    println!("数据后5行：\n{}", df.tail(Some(5)));
    // println!("数据统计：\n{}", df.describe(None));  // describe()方法在describe feature中，要使用 describe() 需要在引入 Polars 时带上 feature。
    println!("获取所有列名：\n{:?}", df.get_column_names());
    println!("获取第一行数据：\n{:?}", df.get_row(0));

    // 处理缺失值，移除 密码 列数据为空的记录
    // let df2 = df.drop_nulls(Some(&["密码".to_string()]));
    // println!("{}", &df2);

    // 填充缺失值，第一个非空值填充后面的空值
    /*
    fill_null提供多种填充策略：
        Forward(Option)：向后遍历，用遇到的第一个非空值（或给定下标位置的值）填充后面的空值
        Backward(Option)：向前遍历，用遇到的第一个非空值（或给定下标位置的值）填充前面的空值
        Mean：用算术平均值填充
        Min：用最小值填充
        Max: 用最大值填充
        Zero：用0填充
        One：用1填充
        MaxBound：用数据类型的取值范围的上界填充
        MinBound：用数据类型的取值范围的下界填充
    */
    // let df3 = df.fill_null(FillNullStrategy::Forward(None));
    // println!("{}", &df3);

    // 去掉重复值，用户 列中重复的数据只会保留第一条
    // drop_duplicates() 接收 2 个参数，第一个参数是个 bool 值，表示是否保持数据的顺序；第二个参数是要处理的列名列表，如果传None表示所有列。
    // let df4 = df.drop_duplicates(true, Some(&["用户".to_string()]))?;
    // println!("{}", &df4);


    // 选择列
    println!("选择列：\n{:?}", df.select(&["用户名".to_string(), "密码".to_string()]));
    println!("用下标获取列：\n{:?}", df.select_by_range(0..2));


    // 数据过滤，筛选角色为 2 的数据
    // const ROLE_COL: &str = "角色";
    // let mask = df.column(ROLE_COL)?.equal(2)?;
    // let filtered = df.filter(&mask)?;
    // println!("角色为 2 的数据：\n{}", filtered);
    Ok(())
}